package org.framework.lazy.cloud.network.heartbeat.protocol.log;

import lombok.extern.slf4j.Slf4j;
import org.framework.lazy.cloud.network.heartbeat.common.advanced.flow.permeate.ChannelFlow;
import org.framework.lazy.cloud.network.heartbeat.protocol.properties.ProtocolProxyProperties;
import org.framework.wu.framework.queue.Message;
import org.framework.wu.framework.queue.MessageQueue;
import org.framework.wu.framework.queue.MessageQueueFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import org.wu.framework.core.utils.ObjectUtils;
import org.wu.framework.lazy.orm.database.lambda.stream.lambda.LazyLambdaStream;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class ProxyReceiverLog implements CommandLineRunner {

    private final ProtocolProxyProperties protocolProxyProperties;

    private final LazyLambdaStream lazyLambdaStream;

    ThreadPoolExecutor CHANNEL_LOG_EXECUTOR =
            new ThreadPoolExecutor(10, 20, 3L, TimeUnit.MINUTES,
                    new LinkedBlockingDeque<>(50), new ThreadPoolExecutor.AbortPolicy());

    public ProxyReceiverLog(ProtocolProxyProperties protocolProxyProperties, LazyLambdaStream lazyLambdaStream) {
        this.protocolProxyProperties = protocolProxyProperties;
        this.lazyLambdaStream = lazyLambdaStream;
    }

    @Override
    public void run(String... args) throws Exception {
        String receiverMsgQueue = protocolProxyProperties.getReceiverMsgQueue();
        MessageQueue receiverQueue = MessageQueueFactory.getQueue(receiverMsgQueue);


        // 创建监听线程
        Thread thread = new Thread(() -> {
            while (true) {
                Message receive = receiverQueue.receive();
                if (ObjectUtils.isNotEmpty(receive)) {
                    ProxyLog proxyLog = (ProxyLog) receive.getBody();
                    lazyLambdaStream.upsert(proxyLog);
                }
            }
        });

        CHANNEL_LOG_EXECUTOR.submit(thread::start);

        String sendMsgQueue = protocolProxyProperties.getSendMsgQueue();
        MessageQueue sendQueue = MessageQueueFactory.getQueue(sendMsgQueue);


        // 创建监听线程
        Thread sendThread = new Thread(() -> {
            while (true) {
                Message receive = sendQueue.receive();
                if (ObjectUtils.isNotEmpty(receive)) {
                    ProxyLog proxyLog = (ProxyLog) receive.getBody();

                    lazyLambdaStream.upsert(proxyLog);
                }
            }
        });

        CHANNEL_LOG_EXECUTOR.submit(sendThread::start);
    }
}
